package health

import (
	"context"
	"gitee.com/zaper/kafka-service/service/connect"
	"gitee.com/zaper/kafka-service/util"
	"gitee.com/zaper/private-logger"
	"github.com/segmentio/kafka-go"
)

func Check(groupIds []string) bool {
	return util.Retry(exec, 2, groupIds)
}

// ctx *context.Context, groupIds []string
func exec(params ...interface{}) bool {
	if len(params) != 1 {
		logger.Info("Input param count is error")
		return false
	}
	groupIds, ok1 := params[0].([]string)
	if !ok1 {
		logger.Info("Input param type is error")
		return false
	}
	gsResp, err := connect.GetKafkaConnectManager().Client.DescribeGroups(context.Background(),
		&kafka.DescribeGroupsRequest{
			GroupIDs: groupIds,
		})
	if err != nil {
		logger.Infof("Get group info fail: %v", err)
		return false
	}
	if len(gsResp.Groups) != len(groupIds) {
		logger.Infof("%s not exist，keep wait ... ", groupIds)
		return false
	}
	for _, group := range gsResp.Groups {
		members := len(group.Members)
		logger.Infof("consumer group %s status: %s, team num: %d.", group.GroupID,
			group.GroupState, members)
		if group.GroupState != "Stable" || members == 0 {
			return false
		}
	}
	logger.Info("All consumer group status ready, go go go!")
	return true
}
